package com.etc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;

/**
 * @Auther: zhezhe
 * @Date: 2018/10/299:44
 * @Description:
 */
public class DataSearch {

    public static class MapDataSearch extends Mapper<LongWritable, Text,Text,Text>{

        private String filename;

        /**
         * 在map之前重写setup方法，读取文件，获得文件名
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {

            FileSplit inputSplit = (FileSplit)context.getInputSplit();
             filename = inputSplit.getPath().getName().toString();
        }

        /**
         * map是对一行数据进行处理
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            //将value转码为GBK格式
            Text gbk = transformTextToUTF8(value, "GBK");
            //读取一行数据
            String line = gbk.toString();
            //以句号切分
            String[] split = line.split("。");
            //如果一行数据中只有一个句号，则不用切分，也不用写循环
            for (String s : split) {
                if (s.contains("月")){
                    context.write(new Text(filename),new Text(s));
                }
            }
        }



        /**
         * 转换编码格式
         * @param text 原数据
         * @param encoding 指定编码格式
         * @return
         */
        public static Text transformTextToUTF8(Text text, String encoding) {
            String value = null;
            try {
                value = new String(text.getBytes(), 0, text.getLength(), encoding);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            return new Text(value);
        }

    }


    public static class ReduceDataSearch extends Reducer<Text,Text,Text,Text>{

        /**
         * 定义一个空字符串line，遍历values，重新给line赋值，将新的line作为value写出
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            String line = "";
            for (Text value : values) {
                line = line + value.toString() + "----";
            }
            context.write(key,new Text(line));

        }
    }

    /**
     *

     /**
     * 用来描述一个作业job
     * 使用哪个mapper类，哪个reducer类，输入文件在哪，输出结果放哪。。。。
     * @param args
     * @throws IOException
     * @throws ClassNotFoundException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //读取配置文件
        Configuration conf = new Configuration();
        //创建一个工作
        Job job = Job.getInstance(conf);

        job.setJarByClass(DataSearch.class);
        job.setMapperClass(MapDataSearch.class);
        job.setReducerClass(ReduceDataSearch.class);


        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

       //如果输出路径已经存在，则将其删除
        File file = new File("E:\\数据检索\\output");
        if (file.exists()){

            FileUtil.fullyDelete(file);
        }
        // 指定要处理的原始数据存放的路径
        FileInputFormat.addInputPath(job,new Path("E:\\数据检索\\input\\*"));
        // 指定处理之后的结果输出到哪个路径
        FileOutputFormat.setOutputPath(job,new Path("E:\\数据检索\\output"));

        //
        job.setNumReduceTasks(1);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
